package com.rrd.mex.dw.mr.babel.parser;

import com.alibaba.fastjson.JSONObject;
import com.rrd.mex.dw.utils.DateUtil;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.mexico.PhoneRiskQueryRecord;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.text.ParsePosition;
import java.util.Date;

/**
 * @Classname PhoneRiskQueryRecordParse
 * @Description Phone Risk Query
 * @Date 2023/8/9 15:11
 * @Created by zxb
 */

public class PhoneRiskQueryRecordParse extends ParseBase {
    Text key = new Text();
    Text value = new Text();

    @Override
    public boolean parseJob(Mapper.Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        PhoneRiskQueryRecord obj = (PhoneRiskQueryRecord) model;
        StringBuilder sb = getBaseInfo(obj, successful,1);

        sb.append(obj.getReferenceId() == null ? "" : obj.getReferenceId()).append(CRT001);
        sb.append(obj.getExternalId() == null ? "" : obj.getExternalId()).append(CRT001);
        sb.append(obj.getValidationCode() == null ? "" : obj.getValidationCode()).append(CRT001);
        sb.append(obj.getMessageCode() == null ? "" : obj.getMessageCode()).append(CRT001);

        JSONObject status = obj.getStatus() == null ? new JSONObject() : obj.getStatus();
        if (status.getString("updated_on") == null){
            sb.append("").append(CRT001);
        } else {
            String dateTime = status.getString("updated_on").replace("T"," ").substring(0,19);
            SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            ParsePosition pos = new ParsePosition(0);
            Date date = df.parse(dateTime,pos);
            sb.append(DateUtil.formateDateStr(new Date(date.getTime() + 8*3600000))).append(CRT001);
        }
        sb.append(formatValue(status.getString("code"))).append(CRT001);
        sb.append(formatValue(status.getString("description"))).append(CRT001);

        JSONObject original = obj.getNumbering().getJSONObject("original") == null ? new JSONObject() : obj.getNumbering().getJSONObject("original");
        sb.append(formatValue(original.getString("country_code"))).append(CRT001);
        sb.append(formatValue(original.getString("complete_phone_number"))).append(CRT001);
        sb.append(formatValue(original.getString("phone_number"))).append(CRT001);

        JSONObject call = obj.getNumbering().getJSONObject("cleansing").getJSONObject("call") == null ? new JSONObject() : obj.getNumbering().getJSONObject("cleansing").getJSONObject("call");
        sb.append(formatValue(call.getString("country_code"))).append(CRT001);
        sb.append(formatValue(call.getString("phone_number"))).append(CRT001);
        sb.append(formatValue(call.getString("cleansed_code"))).append(CRT001);
        sb.append(formatValue(call.getString("min_length"))).append(CRT001);
        sb.append(formatValue(call.getString("max_length"))).append(CRT001);

        JSONObject sms = obj.getNumbering().getJSONObject("cleansing").getJSONObject("sms") == null ? new JSONObject() : obj.getNumbering().getJSONObject("cleansing").getJSONObject("sms");
        sb.append(formatValue(sms.getString("country_code"))).append(CRT001);
        sb.append(formatValue(sms.getString("phone_number"))).append(CRT001);
        sb.append(formatValue(sms.getString("cleansed_code"))).append(CRT001);
        sb.append(formatValue(sms.getString("min_length"))).append(CRT001);
        sb.append(formatValue(sms.getString("max_length"))).append(CRT001);

        JSONObject riskInsights = obj.getRiskInsights() == null ? new JSONObject() : obj.getRiskInsights();
        sb.append(formatValue(riskInsights.getString("status"))).append(CRT001);
        sb.append(formatValue(riskInsights.getString("category"))).append(CRT001);
        sb.append(formatValue(riskInsights.getString("a2p"))).append(CRT001);
        sb.append(formatValue(riskInsights.getString("p2p"))).append(CRT001);
        sb.append(formatValue(riskInsights.getString("number_type"))).append(CRT001);
        sb.append(formatValue(riskInsights.getString("ip"))).append(CRT001);
        sb.append(formatValue(riskInsights.getString("email"))).append(CRT001);

        JSONObject phoneType = obj.getPhoneType() == null ? new JSONObject() : obj.getPhoneType();
        sb.append(formatValue(phoneType.getString("code"))).append(CRT001);
        sb.append(formatValue(phoneType.getString("description"))).append(CRT001);

        JSONObject location = obj.getLocation() == null ? new JSONObject() : obj.getLocation();
        sb.append(formatValue(location.getString("city"))).append(CRT001);
        sb.append(formatValue(location.getString("state"))).append(CRT001);
        sb.append(formatValue(location.getString("zip"))).append(CRT001);
        sb.append(formatValue(location.getString("metro_code"))).append(CRT001);

        JSONObject country = location.getJSONObject("country") == null ? new JSONObject() : location.getJSONObject("country");
        sb.append(formatValue(country.getString("name"))).append(CRT001);
        sb.append(formatValue(country.getString("iso2"))).append(CRT001);
        sb.append(formatValue(country.getString("iso3"))).append(CRT001);

        JSONObject coordinates = location.getJSONObject("coordinates") == null ? new JSONObject() : location.getJSONObject("coordinates");
        sb.append(formatValue(coordinates.getString("latitude"))).append(CRT001);
        sb.append(formatValue(coordinates.getString("longitude"))).append(CRT001);

        JSONObject time_zone = location.getJSONObject("time_zone") == null ? new JSONObject() : location.getJSONObject("time_zone");
        sb.append(formatValue(time_zone.getString("name"))).append(CRT001);
        sb.append(formatValue(time_zone.getString("utc_offset_min"))).append(CRT001);
        sb.append(formatValue(time_zone.getString("utc_offset_max"))).append(CRT001);

        JSONObject carrier = obj.getCarrier() == null ? new JSONObject() : obj.getCarrier();
        sb.append(formatValue(carrier.getString("name"))).append(CRT001);

        JSONObject blocklisting = obj.getBlocklisting() == null ? new JSONObject() : obj.getBlocklisting();
        sb.append(formatValue(blocklisting.getString("blocked"))).append(CRT001);
        sb.append(formatValue(blocklisting.getString("block_code"))).append(CRT001);
        sb.append(formatValue(blocklisting.getString("block_description"))).append(CRT001);

        JSONObject risk = obj.getRisk() == null ? new JSONObject() : obj.getRisk();
        sb.append(formatValue(risk.getString("level"))).append(CRT001);
        sb.append(formatValue(risk.getString("recommendation"))).append(CRT001);
        sb.append(formatValue(risk.getString("score"))).append(CRT001);

        key.set(getKey());
        value.set(sb.toString().replace("\n", " ").replace("\r", " "));
        context.write(key, value);
        return true;
    }

    @Override
    public String getTableName() {
        return "kafka_babel_phone_risk_query_inc";
    }

    @Override
    public String getType() {
        return RecordType.PHONE_RISK_QUERY_RECORD.toString();
    }

    @Override
    public String getComment() {
        return "Phone Risk Query";
    }

    @Override
    public BaseModel getModel() {
        return new PhoneRiskQueryRecord();
    }
}
